package SparkExercises.kejian4;


import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Serializable;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;

public class LianXi3 {
    public static void main(String[] args) {
        /**
         * 自定义一个List<User>类型对象（User含有一个年龄属性），
         * 然后转成合适的键值对RDD，并使用sortByKey算子对这些User按照年龄由大到小排序，如果年龄一样则按分数升序
         * 并将排序后的结果输出到控制台。【注意体会sortBy和sortByKey的区别】
         *
         */
        SparkConf sparkConf = new SparkConf().setAppName("LianXi3").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
//        List<User> users = Arrays.asList(
//                new User("yang", 20, "2020-10"), new User("wang", 39, "2019-10"), new User("yang", 19, "2018-10")
//        );

        List<User> user = new ArrayList<User>();
        User user1 = new User("yang",20,90,"2020-10");
        User user2 = new User("wang",39,80,"2019-10");
        User user3 = new User("yang",19,70,"2018-10");
        User user4 = new User("zang",19,50,"2018-10");
        User user5 = new User("wang",19,60,"2019-10");
        user.add(user1);
        user.add(user2);
        user.add(user3);
        user.add(user4);
        user.add(user5);
        JavaRDD<User> rdd1 = sc.parallelize(user);

        JavaPairRDD<User, User> rdd2 = rdd1.mapToPair(new PairFunction<User, User, User>() {
            @Override
            public Tuple2<User, User> call(User user) throws Exception {
                return new Tuple2<>(user, user);
            }
        });
        JavaPairRDD<User, User> rdd3 = rdd2.sortByKey(new Comp());
        JavaRDD<User> rdd4 = rdd3.values();
        List<User> collect = rdd4.collect();


        for (User u :collect){
            System.out.println(u);
        }

    }
}


class User implements Serializable{
    private String name;
    private Integer age;
    private Integer score;
    private String data;

    public User(String name, Integer age, Integer score, String data) {
        this.name = name;
        this.age = age;
        this.score = score;
        this.data = data;
    }

    public Integer getScore() {
        return score;
    }


    public void setScore(Integer score) {
        this.score = score;
    }

    public User(String name, Integer age, String data) {
        this.name = name;
        this.age = age;
        this.data = data;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }

    public String getData() {
        return data;
    }

    public void setData(String data) {
        this.data = data;
    }

    @Override
    public String toString() {
        return "User{" +
                "name='" + name + '\'' +
                ", age=" + age +
                ", score=" + score +
                ", data='" + data + '\'' +
                '}';
    }


}
class Comp implements Serializable, Comparator<User> {
    @Override
    public int compare(User o1, User o2) {
        if (o1.getAge() > o2.getAge()){
            return 1;
        } else if (o1.getAge() == o2.getAge()){
            if (o1.getScore() > o2.getScore()){
                return 1;
            }else if (o1.getScore() == o2.getScore()){
                return 0;
            }else {
                return -1;
            }
        }else {
            return -1;
        }
    }
}
